package com.bbz.core.lambda.java8.Chapter_11_CompletableFuture;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static java.util.stream.Collectors.toList;

/**
 * Java8 CompletableFuture 实现异步操作
 *
 *  执行比较耗时的操作时，尤其是那些依赖一个或多个远程服务的操作，使用异步任务可
 * 以改善程序的性能，加快程序的响应速度。
 *  你应该尽可能地为客户提供异步API。使用CompletableFuture类提供的特性，你能够
 * 轻松地实现这一目标。
 *  CompletableFuture类还提供了异常管理的机制，让你有机会抛出/管理异步任务执行
 * 中发生的异常。
 *  将同步API的调用封装到一个CompletableFuture中，你能够以异步的方式使用其结果。
 *  如果异步任务之间相互独立，或者它们之间某一些的结果是另一些的输入，你可以将这
 * 些异步任务构造或者合并成一个。
 *  你可以为CompletableFuture注册一个回调函数，在Future执行完毕或者它们计算的
 * 结果可用时，针对性地执行一些程序。
 *  你可以决定在什么时候结束程序的运行，是等待由CompletableFuture对象构成的列表
 * 中所有的对象都执行完毕，还是只要其中任何一个首先完成就中止程序的运行。S
 */

public class Shop {


    private String name;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public Shop(String name) {
        this.name = name;
    }

    static List<Shop> shops = Arrays.asList(new Shop("BestPrice"),
            new Shop("LetsSaveBig"),
            new Shop("MyFavoriteShop"),
            new Shop("BbzShop"),
            new Shop("BbzShop1"),
            new Shop("BbzShop2"),
            new Shop("BbzShop3"),
            new Shop("BbzShop4"),
            new Shop("BbzShop5"),
            new Shop("BbzShop6"),
            new Shop("BbzShop7"),
            new Shop("BuyItAll"));
    private static final Executor executor =Executors.newFixedThreadPool(Math.min(shops.size(), 100),new ThreadFactory() {
                            public Thread newThread(Runnable r) {
                            Thread t = new Thread(r);
                            t.setDaemon(true);
                            return t;
                        }
                    });

    /**
     * java 8 之前使用ExecutorService 实现 future
     *
     * @param product
     * @return
     */
/*    public double getPrice(String product) {
        Double result = 0.0;
        ExecutorService executor = Executors.newCachedThreadPool();
        Future<Double> future = executor.submit(new Callable<Double>() {
            public Double call() {
                return calculatePrice("my favorite product");
            }
        });
        //doSomethingElse();
        try {
            result = future.get(1, TimeUnit.SECONDS);
        } catch (ExecutionException ee) {
            // 计算抛出一个异常
        } catch (InterruptedException ie) {
            // 当前线程在等待过程中被中断
        } catch (TimeoutException te) {
            // 在Future对象完成之前超过已过期
        }
        return result;
    }*/
    public String getPrice(String product) {
        double price = calculatePrice(product);
        Discount.Code code = Discount.Code.values()[
                new Random().nextInt(Discount.Code.values().length)];
        // System.out.println(String.format("%s:%.2f:%s", name, price, code));
        return String.format("%s:%.2f:%s", name, price, code);
    }


    private double calculatePrice(String product) {
        delay();
        return new Random().nextDouble() * product.charAt(0) + product.charAt(1);
    }

    public static void delay() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * java 8 异步操作 使用CompletableFuture
     * 利用completeExceptionally方法可将线程中的异常信息抛出，有利于定位运行中的异常
     * 利用CompletableFuture 工厂方法supplyAsync(),更好更简洁的实现异步操作。
     *
     * @param product
     * @return
     */
    public Future<Double> getPriceAsync(String product) {
        return CompletableFuture.supplyAsync(() -> calculatePrice(product));
     /*   CompletableFuture<Double> completableFuture = new CompletableFuture<>();
        new Thread(() -> {
            try {
                double price = calculatePrice(product);
                completableFuture.complete(price);
            } catch (Exception ex) {
                completableFuture.completeExceptionally(ex);
            }
        }).start();
        return completableFuture;*/
    }

    public static void main(String args[]) {
      /*  Shop shop = new Shop("aa");
        long start = System.nanoTime();
        Future<Double> futurePrice = shop.getPriceAsync("my favorite product");
        long invocationTime = ((System.nanoTime() - start) / 1_000_000);
        System.out.println("Invocation returned after " + invocationTime
                + " msecs");
// 执行更多任务，比如查询其他商店
        //doSomethingElse();
        System.out.println("这个期间可以做任何事情");
// 在计算商品价格的同时
        try {
            double price = futurePrice.get();
            System.out.printf("Price is %.2f%n", price);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        long retrievalTime = ((System.nanoTime() - start) / 1_000_000);
        System.out.println("Price returned after " + retrievalTime + " msecs");


        long start1 = System.nanoTime();
        System.out.println(findPrices("myPhone27S"));
        long duration = (System.nanoTime() - start1) / 1_000_000;
        System.out.println("Done in " + duration + " msecs");*/

      //thenAccept方法可实现有返回值就打印输出
        long start = System.nanoTime();
        CompletableFuture[] futures = findPricesStream("myPhone27S")
                .map(f -> f.thenAccept(
                        s -> System.out.println(s + " (done in " +
                                ((System.nanoTime() - start) / 1_000_000) + " msecs)")))
                .toArray(size -> new CompletableFuture[size]);
        //allOf工厂方法接收一个由CompletableFuture构成的数组，数组中的所有Completable-Future对象执行完成之后，
        // 它返回一个CompletableFuture<Void>对象。这意味着，如果你需要等待最初Stream 中的所有CompletableFuture 对象执行完毕，
        // 对allOf 方法返回的CompletableFuture执行join操作是个不错的主意。
        CompletableFuture.allOf(futures).join();
        System.out.println("All shops have now responded in "
                + ((System.nanoTime() - start) / 1_000_000) + " msecs");
    }

    /**
     * 查找所有商店的商品价格
     *
     * @param product
     * @return
     */
    public static List<String> findPrices(String product) {
        //1:使用stream遍历查找,查询时间5s
        /*return shops.stream()
                .map(shop -> String.format("%s price is %.2f", shop.getName(), shop.getPrice(product)))
                .collect(Collectors.toList());*/
        //2:使用并行流遍历查找,查询时间2s
        /*return shops.parallelStream()
                .map(shop -> String.format("%s price is %.2f", shop.getName(), shop.getPrice(product)))
                .collect(Collectors.toList());*/
        //3:使用CompletableFuture 异步查询时间3s,效果也不理想。
      /* List<CompletableFuture<String>> priceFutures = shops.stream().map(shop -> CompletableFuture.supplyAsync(() ->
                String.format("%s price is %.2f", shop.getName(), shop.getPrice(product))))
                .collect(toList());
        return priceFutures.stream()
                .map(CompletableFuture::join)
                .collect(toList());*/
        //4：使用CompletableFuture 异步查询,并设置线程池大小,效果2s.
        //创建一个线程池，线程池中线程的数目为100和商店数目二者中较小的一个值
        List<CompletableFuture<String>> priceFutures = shops.stream().map(shop ->
                CompletableFuture.supplyAsync(() -> shop.getName() + " price is " + shop.getPrice(product), executor))
                .collect(toList());
        return priceFutures.stream()
                .map(CompletableFuture::join)
                .collect(toList());
    }

    public static Stream<CompletableFuture<String>> findPricesStream(String product) {
        //thenCompose方法允许你对两个异步操作进行流水线，第一个操作完成时，将其结果作为参数传递给第二个操作
       return shops.stream()
                .map(shop -> CompletableFuture.supplyAsync(() -> shop.getPrice(product)))
                .map(future -> future.thenApply(Quote::parse))
                .map(future -> future.thenCompose(quote ->
                        CompletableFuture.supplyAsync(() -> Discount.applyDiscount(quote), executor)));
    }
}
